package cn.com.netedge.iot.frontend.datasave.task;

import cn.com.netedge.iot.common.constants.Globals;
import cn.com.netedge.iot.common.constants.MqConstants;
import cn.com.netedge.iot.common.utils.ChronicleQueueUtil;
import cn.com.netedge.iot.common.utils.StringUtil;
import cn.com.netedge.iot.frontend.datasave.utils.MqConsumeUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

@Component("feDataSaveLocalMessageConsumeTask")
public class LocalMessageConsumeTask {
    private static Logger logger = LoggerFactory.getLogger(LocalMessageConsumeTask.class);

    private static final int INTERVAL = 5 * 1000; // 5s

    private static  List<String> queueNameList = new ArrayList<>(20);
    static  {
        queueNameList.add(MqConstants.QUEUE_DATA_C_ONLINE_MESSAGE);
        queueNameList.add(MqConstants.QUEUE_DATA_C_LOAD_MESSAGE);
        queueNameList.add(MqConstants.QUEUE_DATA_C_LOAD2_MESSAGE);
        queueNameList.add(MqConstants.QUEUE_DATA_C_DAILY_MESSAGE);
        queueNameList.add(MqConstants.QUEUE_DATA_C_MONTHLY_MESSAGE);
        queueNameList.add(MqConstants.QUEUE_DATA_C_EVENT_MESSAGE);
        queueNameList.add(MqConstants.QUEUE_DATA_C_EVENT_COMMUNICATION_MESSAGE);
        queueNameList.add(MqConstants.QUEUE_DATA_C_EVENT_DISCONNECTOR_MESSAGE);
        queueNameList.add(MqConstants.QUEUE_DATA_C_EVENT_FRAUDDETECTION_MESSAGE);
        queueNameList.add(MqConstants.QUEUE_DATA_C_EVENT_POWERFAILURE_MESSAGE);
        queueNameList.add(MqConstants.QUEUE_DATA_C_EVENT_POWERQUALITY_MESSAGE);
        queueNameList.add(MqConstants.QUEUE_DATA_C_EVENT_STANDARD_MESSAGE);
        queueNameList.add(MqConstants.QUEUE_DATA_C_EVENT_VOLTAGEQUALITY_MESSAGE);
    }

    @Scheduled(fixedDelay = INTERVAL)
    public void consume() {
        for(String queueName : queueNameList) {
            int count = 0;
            String message = ChronicleQueueUtil.dequeue(queueName);
            while(StringUtil.isNotEmpty(message) && count < Globals.CHUNK_RECORD_SIZE * 20) {
                count ++;
                int len = message.length();
                if(len > 256) {
                    len = 256;
                }
                logger.debug("consume local message -> queueName:{}, message:{} ", queueName, message.substring(0, len));
                MqConsumeUtil.consume(queueName, message);

                message = ChronicleQueueUtil.dequeue(queueName);
            }
        }
    }
}
